package net.cnki;

import org.apache.http.HttpHost;
import org.elasticsearch.action.DocWriteResponse;
import org.elasticsearch.action.admin.indices.delete.DeleteIndexRequest;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.delete.DeleteRequest;
import org.elasticsearch.action.delete.DeleteResponse;
import org.elasticsearch.action.get.GetRequest;
import org.elasticsearch.action.get.GetResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.support.master.AcknowledgedResponse;
import org.elasticsearch.action.update.UpdateRequest;
import org.elasticsearch.action.update.UpdateResponse;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.client.core.CountRequest;
import org.elasticsearch.client.core.CountResponse;
import org.elasticsearch.client.indices.CreateIndexRequest;
import org.elasticsearch.client.indices.CreateIndexResponse;
import org.elasticsearch.client.indices.GetIndexRequest;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.common.xcontent.XContentBuilder;
import org.elasticsearch.common.xcontent.XContentFactory;
import org.elasticsearch.index.query.BoolQueryBuilder;
import org.elasticsearch.index.query.QueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.index.reindex.BulkByScrollResponse;
import org.elasticsearch.index.reindex.DeleteByQueryRequest;
import org.elasticsearch.rest.RestStatus;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.SearchHits;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.elasticsearch.search.sort.FieldSortBuilder;
import org.elasticsearch.search.sort.ScoreSortBuilder;
import org.elasticsearch.search.sort.SortBuilders;
import org.elasticsearch.search.sort.SortOrder;
import org.elasticsearch.tasks.TaskAwareRequest;

import java.io.IOException;
import java.util.*;

/**
 * @author: Chenzhidiao
 * @date: 2020/4/27 10:11
 * @description:
 * @version: 1.0
 */
public class MyEsUtils {
    //官网指南：https://www.elastic.co/guide/en/elasticsearch/client/java-rest/7.x/java-rest-overview.html

    private static RestHighLevelClient client = null;

    /**
     * 获取客户端
     *
     * @return
     */
    public static RestHighLevelClient getRestHighLevelClient() {
        if (client == null) {
            try {
                client = MyEsClientPool.getClient();
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
        return client;
    }

    /**
     * 关闭客户端
     */
    public static void closeClient() {
        if (client != null) {
            MyEsClientPool.returnClient(client);
        }
    }


    /**
     * 构建查询对象
     *
     * @param fieldsMap 查询条件 (key:查询字段 ,vlues:值)
     * @return
     */
    public static BoolQueryBuilder getQueryBuilder(Map<String, String> fieldsMap) {

        BoolQueryBuilder boolQueryBuilder = QueryBuilders.boolQuery();
        Set<String> strings = fieldsMap.keySet();
        for (String string : strings) {
            //boolQueryBuilder.must(QueryBuilders.wildcardQuery(string, "*" + fieldsMap.get(string) + "*"));
            boolQueryBuilder.must(QueryBuilders.termsQuery("MetadataID",  fieldsMap.get(string) ));
        }
        //boolQueryBuilder.must(QueryBuilders.rangeQuery("id").gte(160000).lte(170000));
        return boolQueryBuilder;
    }

    /**
     * 获取分页后的结果集
     *
     * @param queryBuilder 查询对象
     * @param esIndex      索引名
     * @param pageNo       页数
     * @param pagesize     页大小
     * @return
     */
    public static List<String> getPageResultList(QueryBuilder queryBuilder, String esIndex, int pageNo, int pagesize, String sortField, String[] include) {
        SearchRequest searchRequest = new SearchRequest(esIndex);
        SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();

        if (pageNo <= 1) {
            pageNo = 1;
        }
        //searchSourceBuilder.query(queryBuilder).from((pageNo - 1) * pagesize).size(pagesize).sort(SortBuilders.fieldSort(sortField).order(SortOrder.ASC).unmappedType("long")).fetchSource(include, new String[]{});
        searchSourceBuilder.query(queryBuilder).from((pageNo - 1) * pagesize).size(pagesize).sort(sortField,SortOrder.ASC).fetchSource(include, new String[]{});
        searchRequest.source(searchSourceBuilder);
        client = getRestHighLevelClient();
        SearchResponse searchResponse = null;

        try {
            searchResponse = client.search(searchRequest, RequestOptions.DEFAULT);
        } catch (IOException e) {
            e.printStackTrace();
        }

        // 从response中获得结果
        List<String> list = new LinkedList<String>();
        searchResponse.getHits();

        SearchHits hits = searchResponse.getHits();

        Iterator<SearchHit> iterator = hits.iterator();
        while (iterator.hasNext()) {
            SearchHit next = iterator.next();
            list.add(next.getSourceAsString());
        }
        return list;
    }

    /**
     * 获取结果总数
     *
     * @param queryBuilder
     * @param esIndex
     * @return
     */
    public static Long getResultCount(QueryBuilder queryBuilder, String esIndex) {
        CountRequest countRequest = new CountRequest(esIndex);
        countRequest.query(queryBuilder);
        try {
            CountResponse response = getRestHighLevelClient().count(countRequest, RequestOptions.DEFAULT);
            long length = response.getCount();
            return length;
        } catch (IOException e) {
            e.printStackTrace();
        }
        return 0L;


    }

    /**
     * 获取文档总数
     *
     * @param index
     * @return
     */
    public static long getDocCount(String index) {
        CountRequest countRequest = new CountRequest();
        SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
        searchSourceBuilder.query(QueryBuilders.matchAllQuery());
        countRequest.source(searchSourceBuilder);
        CountResponse countResponse = null;
        client = getRestHighLevelClient();
        try {
            countResponse = client
                    .count(countRequest, RequestOptions.DEFAULT);
        } catch (IOException e) {
            e.printStackTrace();
        }
        return countResponse.getCount();
    }

    /**
     * 判断索引是否存在
     *
     * @param esIndex
     * @return
     */
    public static boolean isIndexExist(String esIndex) {
        boolean isExists = true;
        GetIndexRequest request = new GetIndexRequest(esIndex);
        try {
            isExists = getRestHighLevelClient().indices().exists(request, RequestOptions.DEFAULT);
            if (isExists) {
                System.out.println(String.format("索引%s已存在", esIndex));

            } else {
                System.out.println(String.format("索引%s不存在", esIndex));
            }

        } catch (IOException e) {
            e.printStackTrace();
        }
        return isExists;
    }

    /**
     * 新建索引
     *
     * @param esIndex
     * @param shards       分片数
     * @param replications 副本数
     * @param fields       字段名->类型
     */
    public static void createIndex(String esIndex, int shards, int replications, Map<String, String> fields) {
        if (!isIndexExist(esIndex)) {
            try {
                XContentBuilder builder = XContentFactory.jsonBuilder()
                        .startObject()
                        .field("properties")
                        .startObject();
                for (String s : fields.keySet()) {
                    builder.field(s).startObject().field("index", "true").field("type", fields.get(s)).endObject();
                }
                builder.endObject().endObject();
                CreateIndexRequest request = new CreateIndexRequest(esIndex);
                request.settings(Settings.builder()
                        .put("index.number_of_shards", shards)
                        .put("index.number_of_replicas", replications)
                ).mapping(builder);
                CreateIndexResponse createIndexResponse = getRestHighLevelClient().indices().create(request, RequestOptions.DEFAULT);
                boolean acknowledged = createIndexResponse.isAcknowledged();
                if (acknowledged) {
                    System.out.println(String.format("索引%s创建成功", esIndex));
                } else {
                    System.out.println(String.format("索引%s创建失败", esIndex));
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
        } else {
            System.out.println(String.format("索引%s已存在", esIndex));
        }
    }

    /**
     * 删除索引
     *
     * @param esIndex
     */
    public static void deleteIndex(String esIndex) {
        DeleteIndexRequest request = new DeleteIndexRequest(esIndex);

        try {
            AcknowledgedResponse deleteIndexResponse = getRestHighLevelClient().indices().delete(request, RequestOptions.DEFAULT);
            boolean acknowledged = deleteIndexResponse.isAcknowledged();
            if (acknowledged) {
                System.out.println(String.format("索引%s已删除", esIndex));
            } else {
                System.out.println(String.format("索引%s删除失败", esIndex));
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     * 根据id获取数据，返回map(字段名,字段值)
     *
     * @param esIndex
     * @param id
     * @return
     */
    public static Map<String, Object> getDataById(String esIndex, String id) {
        GetRequest request = new GetRequest(esIndex, id);
        GetResponse response = null;
        Map<String, Object> source = null;
        try {
            response = getRestHighLevelClient().get(request, RequestOptions.DEFAULT);
            if (response.isExists()) {
                source = response.getSource();

            }
        } catch (IOException e) {
            e.printStackTrace();
        }
        return source;


    }

    /**
     * 更新文档
     *
     * @param esIndex
     * @param id
     * @param updateFields 更新的字段名->字段值
     */
    public static void updateDataById(String esIndex, String id, Map<String, Object> updateFields) {
        UpdateRequest request = new UpdateRequest(esIndex, id).doc(updateFields);
        try {
            UpdateResponse response = getRestHighLevelClient().update(request, RequestOptions.DEFAULT);
            if (response.status() == RestStatus.OK) {
                System.out.println(String.format("更新索引为%s,id为%s的文档成功", response.getIndex(), response.getId()));
            } else {
                System.out.println(String.format("更新索引为%s,id为%s的文档失败", response.getIndex(), response.getId()));
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     * 删除指定id的文档
     *
     * @param esIndex
     * @param id
     */
    public static void deleteDataById(String esIndex, String id) {
        DeleteRequest request = new DeleteRequest(esIndex, id);
        try {
            DeleteResponse response = getRestHighLevelClient().delete(request, RequestOptions.DEFAULT);
            if (response.getResult() == DocWriteResponse.Result.DELETED) {
                System.out.println(String.format("id为%s的文档删除成功", id));
            } else {
                System.out.println(String.format("id为%s的文档删除失败", id));
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     * 根据条件删除数据,先根据查询对象获取结果数，结果数/每次处理的最大数=执行删除操作次数
     *
     * @param esIndex
     * @param queryBuilder
     * @return 删除的数据条数
     */
    public static long deleteDataByQuery(String esIndex, QueryBuilder queryBuilder, int maxDocs) {
        long deletedCount = 0L;
        DeleteByQueryRequest request = new DeleteByQueryRequest(esIndex);
        request.setQuery(queryBuilder);
        //最大设置10000
        request.setBatchSize(10000);
        //设置版本冲突时继续
        request.setConflicts("proceed");
        //最多处理文档数
        request.setMaxDocs(maxDocs);
        // 使用滚动参数来控制“搜索上下文”存活的时间
        //request.setScroll(TimeValue.timeValueMinutes(10));
        Long resultCount = getResultCount(queryBuilder, esIndex);
        System.out.println(String.format("待删除数据%d条",resultCount));
        int num = (int) Math.ceil((resultCount / (double) maxDocs));
        try {
            for (int i = 1; i <= num; i++) {
                BulkByScrollResponse bulkByScrollResponse = getRestHighLevelClient().deleteByQuery(request, RequestOptions.DEFAULT);
                deletedCount += bulkByScrollResponse.getDeleted();
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
        return deletedCount;
    }

    /**
     * 批量插入
     *
     * @param esIndex
     * @param datalist 数据集，数据格式为map<字段名,字段值>
     */
    public static void bulkLoad(String esIndex, List<Map<String, Object>> datalist) {
        BulkRequest bulkRequest = new BulkRequest();
        for (Map<String, Object> data : datalist) {
            Object id = data.get("id");
            //如果数据包含id字段，使用数据id作为文档id
            if (id != null) {
                data.remove("id");
                bulkRequest.add(new IndexRequest(esIndex).id(id.toString()).source(data));
            } else {//让es自动生成id
                bulkRequest.add(new IndexRequest(esIndex).source(data));
            }
        }
        try {
            BulkResponse response = getRestHighLevelClient().bulk(bulkRequest, RequestOptions.DEFAULT);
            System.out.println(response.hasFailures());
            if (!response.hasFailures()) {
                System.out.println(String.format("索引%s批量插入成功，共插入%d条", esIndex, datalist.size()));
            } else {
                System.out.println(String.format("索引%s批量插入失败", esIndex));
            }
        } catch (IOException e) {
            e.printStackTrace();
        }

    }

    public static void main(String[] args) {

        Map<String ,String> fieldsMap = new HashMap<String,String>();
        fieldsMap.put("IndicatorName","农业总产值-牧业产值-");
        fieldsMap.put("MetadataID","T_8_历年农业总产值及指数");
        BoolQueryBuilder queryBuilder = getQueryBuilder(fieldsMap);
        /*long deleteCount = deleteDataByQuery("nv_excel_pickup_origi_new", queryBuilder,10000);
        System.out.println(String.format("删除数据%d条",deleteCount));*/

        String[] include = {"Value","MetadataID","ID","CreateTime","EsTime","UpadteTime"};
        List<String> list = getPageResultList(queryBuilder, "nv_excel_pickup",1,10,"adg",include);
       /* System.out.println(list.size());
        for (String s : list) {
            System.out.println(s);
        }*/
        //3000443-81714-79296-45412
        System.out.println("count:"+getResultCount(queryBuilder,"nv_excel_pickup"));
        /*HashMap<String, String> fields = new HashMap<String, String>();
        fields.put("name","keyword");
        fields.put("age","long");
        //fields.put("create_time","keyword");

        createIndex("company3",1,0,fields);*/
        //deleteIndex("company2");
       /* HashMap<String, Object> fields = new HashMap<String, Object>();
        fields.put("age",25);
        fields.put("name","wangwu2");
        fields.put("@version",3);*/


        //deleteDataById("company","4");
      /*  ArrayList<Map<String, Object>> datas = new ArrayList<Map<String, Object>>();
        HashMap<String, Object> data1 = new HashMap<String, Object>();
        data1.put("id",5);
        data1.put("name","zhangsan");
        data1.put("age",18);
        HashMap<String, Object> data2 = new HashMap<String, Object>();
        data2.put("id",7);
        data2.put("name","diao");
        data2.put("age",22);
        datas.add(data1);
        datas.add(data2);
        bulkLoad("company",datas);*/
        //closeClient();
    }

}
